// -*-c++-*-

#ifndef _TABLET_HPP_
#define _TABLET_HPP_

#include <map>
#include <vector>
#include <ext/hash_map>

#include <boost/pool/pool_alloc.hpp>

#include <constants.hpp>
#include <protocol.hpp>
#include <xbase_utility.hpp>
#include <atomic_adapter.hpp>
#include <logger.hpp>
#include <row_mutation_data.hpp>
#include <performance_measurement.hpp>

#include "table_schema.hpp"
#include "sstable.hpp"
#include "memory_table.hpp"
#include "log.hpp"
#include "naming_strategy.hpp"

#include "group.hpp"
#include "group_scanner.hpp"
#include "service.hpp"

namespace xbase {
	class tablet_scanner;	// prototype
	typedef boost::shared_ptr<tablet_scanner> tablet_scanner_pointer;
	
	// tablet - 子表
	class tablet
		: public boost::enable_shared_from_this<tablet>
	{
		friend class tablet_scanner;
	public:
		//typedef stringable_tag stringable_tag;

		typedef boost::shared_ptr<tablet> pointer;
		typedef __gnu_cxx::hash_map<std::string, group::pointer> group_map;

	public:
		static
		pointer load (const std::string &p_root_path,
			      const tablet_info &p_info);

		~tablet() {
			assert(m_scanner_count == 0);
		}

		// 
		// 读写接口
		// 
	
		// 写入一条记录，到相应的组内
		//
		// 统一在子表级别，按组分类的时候，保证时间戳都设置好了，在组及以下级别中就不再检查了
		// int add(const boost::asio::const_buffer &data,
		// 	const uint64_t &timestamp,
		// 	request::pointer req_ptr,
		// 	uint32_t &error_code);
		int add(const protocol::row_mutation_data_for_tabletserver &data,
			const boost::asio::const_buffer &raw_data,
			error::error_code &error_code);
		
		// 返回一个包含所有组的扫描器的组合扫描器
		tablet_scanner_pointer get_scanner (kv_scan_context::pointer);

		size_t group_number() const {
			return m_groups.size();
		}

		std::string start_row() const {
			return m_start_row;
		}
		
		std::string end_row() const {
			return m_end_row;
		}

		std::string to_string() const {
			return m_tablet_name;
		}

		std::string tablet_name() const {
			return m_tablet_name;
		}

		std::string table_name() const {
			return m_table_name;
		}
		
		bool major_compact_or_split();

		void notify_minor_compact();
		
		void start_minor_compact() {
			boost::mutex::scoped_lock lock(m_compact_status_lock);
			LOG_IF((m_compact_status != cs_minor_compact_reported),
			       WARN,
			       "tablet::start_minor_compact status error" << m_compact_status);
			m_compact_status = cs_doing_minor_compact;
		}

		void complete_minor_compact() {
			boost::mutex::scoped_lock lock(m_compact_status_lock);
			LOG_IF((m_compact_status != cs_doing_minor_compact),
			       WARN,
			       "tablet::complete_minor_compact status error" << m_compact_status);
			m_compact_status = cs_null;
		}

		void notify_major_compact();

		void start_major_compact() {
			boost::mutex::scoped_lock lock(m_compact_status_lock);
			LOG_IF((m_compact_status != cs_major_compact_reported),
			       WARN,
			       "tablet::start_major_compact status error" << m_compact_status);
			m_compact_status = cs_doing_major_compact;
		}

		void complete_major_compact() {
			boost::mutex::scoped_lock lock(m_compact_status_lock);
			LOG_IF((m_compact_status != cs_doing_major_compact),
			       WARN,
			       "tablet::complete_major_compact status error " << m_compact_status);
			m_compact_status = cs_null;
		}

		void check_compact();
		
		// 用于分裂子表后，提交失败时，清空分裂生成的目录。
		//
		// 要避免删错目录！
		void clean_split_path() const;

		bool is_split() const {
			return TabletStatusGetSplitting(this->m_status);
		}
		
		bool is_online() const {
			return TabletStatusGetOnline(this->m_status);
		}

		bool is_deleted() const {
			return TabletStatusGetDeleted(this->m_status);
		}

		bool is_root_meta_tablet() const {
			return m_tablet_id == 0;
		}

		protocol::tablet_info get_tablet_info() const {
			protocol::tablet_info _info(m_tablet_id,
						    m_table_name,
						    m_start_row,
						    m_end_row,
						    m_status);
			return _info;
		}
		protocol::tablet_info get_split_a_info() const {
			protocol::tablet_info _info(m_split_a_id,
						    m_table_name,
						    m_start_row,
						    m_split_row,
						    TABLET_STATUS_CLEAR);
			return _info;
		}
		protocol::tablet_info get_split_b_info() const {
			protocol::tablet_info _info(m_split_b_id,
						    m_table_name,
						    m_split_row,
						    m_end_row,
						    TABLET_STATUS_CLEAR);
			return _info;
		}

		bool do_minor_compact_forced() {
			bool need_major_compact;
			return minor_compact(true, need_major_compact);
		}
		bool do_minor_compact(bool &need_major_compact) {
			return minor_compact(false, need_major_compact);
		}
		
	protected:
		bool minor_compact(bool p_forced, bool &need_major_compact);

		bool is_need_minor_compact() const {
			boost::shared_lock<boost::shared_mutex> guard(m_groups_lock);
			BOOST_FOREACH(const group_map::value_type &elem, m_groups)
			{
				if(elem.second->need_minor_compact())
					return true;
			}
			return false;
		}

		bool is_need_major_compact() const {
			boost::shared_lock<boost::shared_mutex> guard(m_groups_lock);
			BOOST_FOREACH(const group_map::value_type &elem, m_groups)
			{
				if(elem.second->need_major_compact())
					return true;
			}
			return false;
		}
		
	public:

		uint64_t tablet_id() const {
			return m_tablet_id;
		}

		uint64_t max_sequence_number() const;

		uint32_t memory_table_size() const {
			uint32_t size = 0;
			boost::shared_lock<boost::shared_mutex> guard(m_groups_lock);
			BOOST_FOREACH(const group_map::value_type &elem, m_groups)
			{
				size += elem.second->memory_table_size();
			}
			return size;
		}

		boost::shared_mutex &lock() {
			return m_lock;
		}

		uint32_t get_scanner_count() const {
			return m_scanner_count.value();
		}

		// 暂时，在获取写锁的情况下，没有扫描器存在即是子表没有
		// 使用
		bool not_used() const {
			return (m_scanner_count == 0);
		}

		uint16_t set_split(bool p_val) {
			uint16_t _status = m_status;
			TabletStatusSetSplitting(this->m_status, p_val);
			return _status;
		}

		uint16_t set_online(bool p_val) {
			uint16_t _status = m_status;
			TabletStatusSetOnline(this->m_status, p_val);
			return _status;
		}

		uint16_t set_deleted(bool p_val) {
			uint16_t _status = m_status;
			TabletStatusSetDeleted(this->m_status, p_val);
			return _status;
		}

	protected:
		tablet(uint64_t p_tablet_id,
		       const std::string p_table_name,
		       const std::string p_start_row,
		       const std::string p_end_row,
		       uint16_t p_status);

		table_schema::pointer get_schema() const; // TODO: inline it

		void split_row(const std::string &p_split_row) {
			this->m_split_row = p_split_row;
			this->m_split_a_id = cur_micro_seconds();
			const uint64_t mask = 0x1LL;
			// 注意：将最后一位置为0，使得偶数总为上半部
			// 分，奇数总为下半部分。SSTable中就是根据这个
			// 特征来判断是哪半部分的。
			this->m_split_a_id &= ~mask;
			this->m_split_b_id = this->m_split_a_id + 1;
		}
		
		std::string split_row() const {
			return this->m_split_row;
		}

		std::string split_a_tablet_name() const {
			return naming_strategy::tablet_name(m_split_a_id,
							    m_table_name,
							    m_start_row,
							    m_split_row);
		}
		std::string split_b_tablet_name() const {
			return naming_strategy::tablet_name(m_split_b_id,
							    m_table_name,
							    m_split_row,
							    m_end_row);
		}

		bool split();

		void append(const group::pointer &p_group);
		
		group::pointer get_or_add_group(const std::string &group);

		bool loaded(const std::string &p_groupname) const {
			boost::shared_lock<boost::shared_mutex> guard(m_groups_lock);
			return m_groups.find(p_groupname) != m_groups.end();
		}

		// 专门用于恢复后的紧缩
		bool recover_compact();
		
		bool load_oldlog(const std::string &p_path);

		void inc_scanner_count() {
			assert(m_scanner_count >= 0);
			++ m_scanner_count;
		}
		void dec_scanner_count() {
			-- m_scanner_count;
			assert(m_scanner_count >= 0);
		}

	private:
		tablet(const tablet &);
		tablet &operator=(const tablet &);

	private:
		group_map m_groups;
		// 子表上线后，通过该状态标志来协调紧缩操作；下线后就不需要
		// 0 - none/done
		// 1 - minor compact reported
		// 2 - doing minor compact
		// 3 - major compact reported
		// 4 - doing major compact
		enum compact_status {
			cs_null = 0,
			cs_minor_compact_reported,
			cs_doing_minor_compact,
			cs_major_compact_reported,
			cs_doing_major_compact // 包含了split
		};
		
		compact_status m_compact_status;
		boost::mutex m_compact_status_lock;
		mutable boost::shared_mutex m_groups_lock;

		uint64_t m_tablet_id;
		std::string m_table_name;
		std::string m_start_row;
		std::string m_end_row;
		uint16_t m_status;

		atomic_adapter<uint32_t> m_scanner_count;
		bool m_is_compacting;
		boost::mutex m_is_compacting_lock;

		std::string m_tablet_name;

		std::string m_split_row; // 分裂行，由此可以得到两个新的子表；为空表示没有分裂。
		uint64_t m_split_a_id;
		uint64_t m_split_b_id;

		// 子表的读写锁， 定义子表的读、写、扫描、次要紧缩等不
		// 改变子表状态的操作为读操作，并且可以同时执行，互不影
		// 响；定义主要紧缩、下线子表等为写操作，因为需要关闭子
		// 表，不能再执行上面的任何读操作。
		//
		// 为子表做任何操作，都必须首先获取该锁，再获取其它锁，
		// 避免死锁！
		mutable boost::shared_mutex m_lock;
		
		mutable table_schema::weak_pointer m_schema; // pointer cache
		mutable boost::mutex m_schema_lock;
	};

	class tablet_scanner
		: public kv_scanner
	{
	public:
		struct greater_compare {
			inline
			bool operator() (const group_scanner::pointer &first,
					 const group_scanner::pointer &second) const {
				key_less_than key_lt;
				// second.entry.key < first.entry.key => first > second
				return key_lt(*second->get().m_key,
					      *first->get().m_key);
			}
		};

	public:
		typedef boost::shared_ptr<tablet_scanner> pointer;

		tablet_scanner(const kv_scan_context::pointer &context,
			       const tablet::pointer &tablet)
			: kv_scanner(context)
			, m_tablet(tablet) {
			m_tablet->inc_scanner_count();
		}

		virtual
		~tablet_scanner() {
			m_tablet->dec_scanner_count();

			// 注意：
			// 必须先清空队列，再释放子表指针。因为发现当扫
			// 描器超时且队列中仍有内容、子表只剩下这一个
			// 时，先释放子表指针会使组析构，使得缓存管理单
			// 元被释放，但这时仍有块在使用中，所以析构缓存
			// 管理单元时就会core掉。
			while(! m_queue.empty())
			{
				m_queue.pop();
			}
			m_tablet = tablet::pointer();
		}

		virtual
		bool eos() const {
			return m_queue.empty();
		}

		virtual
		const block_entry &get() const {
			assert(! m_queue.empty());
			return m_queue.top()->get();
		}

		// 直接调用第一个组的next即可
		virtual
		void next() {
			if(! m_queue.empty())
			{
				boost::mutex::scoped_lock guard(m_lock);
				next_no_lock();
			}
		}

		void add(const group_scanner::pointer &scanner) {
			const bool eos = scanner->eos();
			if(! eos)
			{
				m_queue.push(scanner);
			}
			// else drop it
		}

		void fill_records_by_row(RecordVector &records) {
			MEASUREMENT_OF_TIME("tablet_scanner::fill_records_by_row", this->id());

			if(m_queue.empty())
				return;

			boost::mutex::scoped_lock guard(m_lock);

			if(m_queue.empty()) // double check: 下面this->get()需要有这个保证
				return;

			const int row_number = scan_context_ptr()->num_row();
			std::string cur_row = (std::string)*(this->get().m_key->row());
			
			int row_count = 0;
			Record record;
			for(; ! m_queue.empty(); this->next_no_lock())
			{
				const block_entry &entry = this->get();
				if(cur_row != *(entry.m_key->row())) // row changed
				{
					++ row_count;
					if(row_count >= row_number) // enough
					{
						break;
					}
					else // need read next row
					{
						cur_row = (std::string)*(entry.m_key->row());
					}
				}
				record.key.row_name = (std::string)*entry.m_key->row();
				record.key.column_name = (std::string)*entry.m_key->column();
				record.key.time_stamp = entry.m_key->timestamp()->get();
				// 如果客户端不想要值的话，就不返回给他
				if(scan_context_ptr()->filter_value_disabled())
				{
					record.value = (std::string)*entry.m_value;
				}
				else
				{
					record.value.clear();
				}

				records.push_back(record);
			}
		}

		bool empty() const {
			return m_queue.empty();
		}

		size_t size() const {
			return m_queue.size();
		}

	protected:
		void next_no_lock() {
			MEASUREMENT_OF_TIME("tablet_scanner::next_no_lock", this->id());
			if(! m_queue.empty())
			{
				group_scanner::pointer scanner = m_queue.top();
				m_queue.pop();
				scanner->next();
				const bool eos = scanner->eos();
				if(! eos)
				{
					m_queue.push(scanner);
				}
			}
		}

	private:
		std::priority_queue<group_scanner::pointer,
				    std::vector<group_scanner::pointer>,
				    greater_compare> m_queue;
		boost::mutex m_lock;
		tablet::pointer m_tablet;
	};

} // namespace xbase

#endif	// _TABLET_HPP_
